package com.faithbj.process.impl;

import com.faithbj.process.*;
import com.faithbj.support.fp.Tuple2;
import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.*;
import org.drools.runtime.StatefulKnowledgeSession;
import org.drools.runtime.process.ProcessInstance;
import org.jbpm.process.audit.JPAWorkingMemoryDbLogger;
import org.jbpm.process.workitem.wsht.HornetQHTWorkItemHandler;
import org.jbpm.task.Content;
import org.jbpm.task.Task;
import org.jbpm.task.TaskData;
import org.jbpm.task.TaskService;
import org.jbpm.task.query.TaskSummary;
import org.jbpm.task.service.ContentData;
import org.jbpm.task.service.FaultData;
import org.jbpm.task.service.SyncTaskServiceWrapper;
import org.jbpm.task.service.hornetq.AsyncHornetQTaskClient;
import org.jbpm.task.utils.ContentMarshallerContext;
import org.jbpm.task.utils.ContentMarshallerHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.DependsOn;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcDaoSupport;
import org.springframework.oxm.jaxb.Jaxb2Marshaller;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.Query;
import javax.sql.DataSource;
import javax.xml.transform.stream.StreamSource;
import java.io.IOException;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;

import static com.faithbj.process.TaskUtils.getText;

/**
 * User: xiongyuanwei
 * Date: 12-7-4
 * Time: 下午9:09
 */
@SuppressWarnings("ALL")
@Service
@DependsOn("taskServer")
public class ProcessServiceImpl extends NamedParameterJdbcDaoSupport implements ProcessEngineService, ProcessQueryService, TaskContentProvider, ApplicationContextAware {

    private static final String MARSHALLING_STRATEGY_TYPE = "org.drools.marshalling.impl.SerializablePlaceholderResolverStrategy";

    private static Logger log = LoggerFactory.getLogger(ProcessServiceImpl.class);

    @Autowired
    private StatefulKnowledgeSession ksession;

    private JPAWorkingMemoryDbLogger jbpmLogger;

    private TaskService taskClient;

    private HornetQHTWorkItemHandler hornetQHTWorkItemHandler;

    @Autowired
    private Jaxb2Marshaller marshaller;

    private ApplicationContext applicationContext;

    private Map<String, NamedQueries.NamedQuery> entityMappings = Maps.newHashMap();

    @Autowired
    private ContentSearcherFactory contentSearcherFactory;

    @Autowired
    @Qualifier("processEmf")
    private EntityManagerFactory processEmf;

    private EntityManager processEm;

    private List<String> groups = Arrays.asList("outer", "inner");

    private Function<Task, TaskSummary> taskToSummaryFn =
            new Function<Task, TaskSummary>() {
                @Override
                public TaskSummary apply(Task task) {
                    TaskData data = task.getTaskData();
                    return new TaskSummary(task.getId(), data.getProcessInstanceId(), getText(task.getNames()),
                            getText(task.getSubjects()), getText(task.getDescriptions()), data.getStatus(),
                            task.getPriority(), data.isSkipable(), data.getActualOwner(), data.getCreatedBy(),
                            data.getCreatedOn(), null, data.getExpirationTime(), data.getProcessId(),
                            data.getProcessSessionId());
                }
            };

    private Function<TaskSummary, TaskModel> taskSummaryToModelFn =
            new Function<TaskSummary, TaskModel>() {
                @Override
                public TaskModel apply(TaskSummary taskSummary) {
                    return new TaskModel(taskSummary.getId(), taskSummary.getName(),
                            taskSummary.getActualOwner() != null ? taskSummary.getActualOwner().getId() : null,
                            taskSummary.getCreatedOn(), taskSummary.getProcessInstanceId(), null);
                }
            };

    private Function<TaskModel, TaskModel> taskModelWithContentFn =
            new Function<TaskModel, TaskModel>() {
                @Override
                public TaskModel apply(TaskModel task) {
                    Map<String, Object> content = getContentByTaskId(task.getId());
                    return new TaskModel(task.getId(), task.getName(), task.getUser(), task.getCreatedOn(), task.getProcessInstanceId(), content);
                }
            };

    @PostConstruct
    protected void init() throws IOException {
        hornetQHTWorkItemHandler = new HornetQHTWorkItemHandler(ksession);
        jbpmLogger = new JPAWorkingMemoryDbLogger(ksession);
        ksession.getWorkItemManager().registerWorkItemHandler("Human Task", hornetQHTWorkItemHandler);
        taskClient = new SyncTaskServiceWrapper(new AsyncHornetQTaskClient("Dana Human Task Client"));
        taskClient.connect();
        NamedQueries namedQueries = (NamedQueries) marshaller.unmarshal(new StreamSource(
                applicationContext.getResource("classpath:META-INF/named-queries.xml").getInputStream()));
        for (NamedQueries.NamedQuery namedQuery : namedQueries.getQueries()) {
            entityMappings.put(namedQuery.getName(), namedQuery);
        }
        processEm = processEmf.createEntityManager();
    }

    @PreDestroy
    protected void destroy() {
        try {
            taskClient.disconnect();
            hornetQHTWorkItemHandler.dispose();
        } catch (Exception e) {
            // nothing to do
        }
        jbpmLogger.dispose();
        ksession.dispose();
    }

    @Override
    public long startProcess(Map<String, Object> vars) {
        ProcessInstance processInstance = ksession.startProcess("com.faithbj.dana.process", Maps.newHashMap(vars));
        return processInstance.getId();
    }

    @Override
    public void claim(long taskId, String userId, TaskModelFilter filter) {
        if (filter.accepts(getTaskWithContent(taskId))) {
            taskClient.claim(taskId, userId);
        } else {
            log.warn("User {} does not have permission to claim task #{}", userId, taskId);
            throw new AccessDeniedException("No permission to claim task.");
        }
    }

    @Override
    public void claim(long taskId, String userId, List<String> groups, TaskModelFilter filter) {
        if (filter.accepts(getTaskWithContent(taskId))) {
            taskClient.claim(taskId, userId, groups);
        } else {
            log.warn("User {} does not have permission to claim task #{}", userId, taskId);
            throw new AccessDeniedException("No permission to claim task.");
        }
    }

    @Override
    public void start(long taskId, String userId) {
        taskClient.start(taskId, userId);
    }

    @Override
    public void complete(long taskId, String userId) {
        complete(taskId, userId, Maps.<String, Object>newHashMap());
    }

    @Override
    public void complete(long taskId, String userId, Map<String, Object> output) {
        ContentData content = output == null ? null :
                ContentMarshallerHelper.marshal(Maps.newHashMap(output), getMarshallerContext(), null);
        taskClient.complete(taskId, userId, content);
    }

    @Override
    public void fail(long taskId, String userId, String fault) {
        FaultData faultData = new FaultData();
        faultData.setFaultName(fault);
        taskClient.fail(taskId, userId, faultData);
    }

    @Override
    public TaskModel getTaskWithContent(long taskId) {
        return taskModelWithContentFn.apply(getTask(taskId));
    }

    @Override
    public List<TaskModel> listAvailableTasksWithContentByUser(String userId, int first, int limit, final TaskModelFilter filter) {
        List<TaskModel> tasks = listAvailableTasksByUser(userId, first, limit);
        List<TaskModel> taskModels = Lists.transform(tasks, taskModelWithContentFn);
        Collection<TaskModel> filtered = Collections2.filter(taskModels, new Predicate<TaskModel>() {
            @Override
            public boolean apply(TaskModel task) {
                return filter.accepts(task);
            }
        });
        return Lists.newArrayList(filtered);
    }

    @Override
    public List<TaskModel> listAllAvailableTasksWithContent(int first, int limit) {
        List<TaskModel> tasks = listAllAvaiableTasks(first, limit);
        List<TaskModel> taskModels = Lists.transform(tasks, taskModelWithContentFn);
        return Lists.newArrayList(taskModels);
    }

    @Override
    public int countAllAvailableTasks() {
        return ((Number) taskClient.query(entityMappings.get("CountTasksAssignedAsBusinessAdministrator").applyParams(
                ImmutableMap.<String, String>of("userId", wrap("Administrator"))), 1, 0).get(0)).intValue();
    }

    @Override
    public List<TaskModel> listAvailableTasksWithContentByUserAndState(String userId, State state, int first, int limit, final TaskModelFilter filter) {
        List<TaskModel> tasks = listAvaiableTasksByUserAndState(userId, state, first, limit);
        List<TaskModel> tasksWithContent = Lists.transform(tasks, taskModelWithContentFn);
        Collection<TaskModel> filtered = Collections2.filter(tasksWithContent, new Predicate<TaskModel>() {
            @Override
            public boolean apply(TaskModel task) {
                return filter.accepts(task);
            }
        });
        return Lists.newArrayList(filtered);
    }

    @Override
    public int countAvailableTasksByUserAndState(String userId, State state, TaskModelFilter filter) {
        Query query = processEm.createNamedQuery("CountTasksAssignedAsPotentialOwnerByState");
        query.setParameter("userId", userId);
        query.setParameter("groups", groups);
        query.setParameter("taskName", state.getName());
        return ((Number) query.getSingleResult()).intValue();
    }

    @Override
    public List<TaskModel> listAllAvailableTasksWithContentByState(State state, int first, int limit) {
        List<TaskModel> tasks = listAllAvaiableTasksByState(state, first, limit);
        return Lists.newArrayList(Lists.transform(tasks, taskModelWithContentFn));
    }

    @Override
    public int countAllAvailableTasksByState(State state) {
        return ((Number) taskClient.query(entityMappings.get("CountTasksAssignedAsBusinessAdministratorByState").applyParams(
                ImmutableMap.<String, String>of("userId", wrap("Administrator"), "taskName", wrap(state.getName()))), 0, 1).get(0)).intValue();
    }

    @Override
    public <T> List<TaskModel> search(Condition<T> condition, String userId, State state, int first, int limit,
                                      Function<Map<String, Object>, T> extractFn, TaskContentFilter filter) {
        return contentSearcherFactory.createContentSearcher(condition, userId, state, first, limit, extractFn, filter)
                .search().getResults();
    }

    @Override
    public <T> int countSearch(Condition<T> condition, String userId, State state, Function<Map<String, Object>, T> extractFn, TaskContentFilter filter) {
        return contentSearcherFactory.createContentSearcher(condition, userId, state, 0, Integer.MAX_VALUE, extractFn, filter).countSearch();
    }

    @Override
    public <T> List<TaskModel> searchAll(Condition<T> condition, State state, int first, int limit, Function<Map<String, Object>, T> extractFn) {
        return search(condition, "Administrator", state, first, limit, extractFn, DefaultTaskModelFilter.getInstance());
    }

    @Override
    public <T> Multimap<Object, TaskModel> groupedSearchAll(Condition<T> condition, State state, int first, int limit, Function<Map<String, Object>, T> extractFn) {
        return contentSearcherFactory.createContentSearcher(condition, "Administrator", state, first, limit, extractFn, DefaultTaskModelFilter.getInstance())
                .search().getGroupedResults();
    }

    @Override
    public long getTaskIdByProcessInstanceIdAndState(long processInstanceId, State state) {
        return ((Number) taskClient.query(entityMappings.get("TaskIdByProcessInstanceIdAndState").applyParams(
                ImmutableMap.<String, String>of("processInstanceId", String.valueOf(processInstanceId),
                        "taskName", wrap(state.getName()))), 1, 0).get(0)).longValue();
    }

    @Override
    public Map<Long, StartAndEndDateOfTaskTuple> getStartAndEndDatesByTasks(List<Long> taskIds) {
        if (taskIds.isEmpty()) return Collections.emptyMap();
        Query query = processEm.createNamedQuery("ProcessInstanceIdNamesByTaskIds");
        query.setParameter("taskIds", taskIds);
        List<TaskIdNameProcessInstanceIdTuple> tuples = query.getResultList();
        final Map<Long, Long> map = Maps.newHashMap();
        final Map<Tuple2<Long, String>, Long> map2 = Maps.newHashMap();
        Set<String> nodeNames = Sets.newHashSet();
        for (TaskIdNameProcessInstanceIdTuple t : tuples) {
            map.put(t.getTaskId(), t.getProcessInstanceId());
            map2.put(new Tuple2<Long, String>(t.getProcessInstanceId(), t.getTaskName()), t.getTaskId());
            nodeNames.add(StateFactory.getStateByTaskName(t.getTaskName()).getNodeName());
        }
        Collection processInstanceIds = Sets.newHashSet(map.values());
        Map<Long, StartAndEndDateOfTaskTuple> ret = Maps.newHashMap();
        if (processInstanceIds.isEmpty() || nodeNames.isEmpty()) return ret;
        Map<String, Object> params = ImmutableMap.<String, Object>of("processInstanceIds", processInstanceIds, "nodeNames", nodeNames);
        List<ProcessInstanceIdNodeNameStartAndEndDateTuple> tuples2 = getNamedParameterJdbcTemplate().query(
                "select n1.processinstanceid, n1.nodename, n1.log_date, n2.log_date from nodeinstancelog n1 left join nodeinstancelog n2 " +
                        "on n1.processinstanceid = n2.processinstanceid and n1.nodeinstanceid = n2.nodeinstanceid and n2.type = 1 " +
                        "where n1.processinstanceid in (:processInstanceIds) and n1.nodename in (:nodeNames) and n1.type = 0",
                params,
                new RowMapper<ProcessInstanceIdNodeNameStartAndEndDateTuple>() {
                    @Override
                    public ProcessInstanceIdNodeNameStartAndEndDateTuple mapRow(ResultSet resultSet, int i) throws SQLException {
                        ProcessInstanceIdNodeNameStartAndEndDateTuple ret = new ProcessInstanceIdNodeNameStartAndEndDateTuple();
                        ret.setProcessInstanceId(resultSet.getLong(1));
                        ret.setNodeName(resultSet.getString(2));
                        ret.setStartDate(resultSet.getDate(3));
                        ret.setEndDate(resultSet.getDate(4));
                        return ret;
                    }
                });
        for (ProcessInstanceIdNodeNameStartAndEndDateTuple t : tuples2) {
            String taskName = StateFactory.getStateByNodeName(t.getNodeName()).getName();
            Tuple2<Long, String> t2 = new Tuple2<Long, String>(t.getProcessInstanceId(), taskName);
            if (map2.containsKey(t2)) {
                long taskId = map2.get(t2);
                StartAndEndDateOfTaskTuple tuple = new StartAndEndDateOfTaskTuple(taskId, t.getStartDate(), t.getEndDate());
                ret.put(taskId, tuple);
            }
        }
        return ret;
    }

    @Override
    public Table<Long, State, Tuple2<Date, Date>> getStartAndEndDatesByProcessInstancesAndStates(Collection<Long> processInstanceIds, Collection<String> nodeNames) {
        Table<Long, State, Tuple2<Date, Date>> ret = HashBasedTable.create();
        if (processInstanceIds.isEmpty() || nodeNames.isEmpty()) return ret;
        Map<String, Object> params = ImmutableMap.<String, Object>of("processInstanceIds", processInstanceIds, "nodeNames", nodeNames);
        List<ProcessInstanceIdNodeNameStartAndEndDateTuple> tuples = getNamedParameterJdbcTemplate().query(
                "select n1.processinstanceid, n1.nodename, n1.log_date, n2.log_date from nodeinstancelog n1 left join nodeinstancelog n2 " +
                        "on n1.processinstanceid = n2.processinstanceid and n1.nodeinstanceid = n2.nodeinstanceid and n2.type = 1 " +
                        "where n1.processinstanceid in (:processInstanceIds) and n1.nodename in (:nodeNames) and n1.type = 0",
                params,
                new RowMapper<ProcessInstanceIdNodeNameStartAndEndDateTuple>() {
                    @Override
                    public ProcessInstanceIdNodeNameStartAndEndDateTuple mapRow(ResultSet resultSet, int i) throws SQLException {
                        ProcessInstanceIdNodeNameStartAndEndDateTuple ret = new ProcessInstanceIdNodeNameStartAndEndDateTuple();
                        ret.setProcessInstanceId(resultSet.getLong(1));
                        ret.setNodeName(resultSet.getString(2));
                        ret.setStartDate(resultSet.getDate(3));
                        ret.setEndDate(resultSet.getDate(4));
                        return ret;
                    }
                });
        for (ProcessInstanceIdNodeNameStartAndEndDateTuple t : tuples) {
            ret.put(t.getProcessInstanceId(), StateFactory.getStateByNodeName(t.getNodeName()), new Tuple2<Date, Date>(t.getStartDate(), t.getEndDate()));
        }
        return ret;
    }

    @Override
    public Multimap<Long, TaskModel> getTaskModelsByProcessInstanceIds(List<Long> processInstanceIds) {
        if (processInstanceIds.isEmpty()) return HashMultimap.create();
        Query query = processEm.createNamedQuery("TasksByProcessInstanceIds");
        query.setParameter("processInstanceIds", processInstanceIds);
        List<TaskIdNameProcessInstanceIdTuple> tuples = query.getResultList();
        Multimap<Long, TaskModel> ret = LinkedListMultimap.create();
        for (TaskIdNameProcessInstanceIdTuple t : tuples) {
            TaskModel task = getTaskWithContent(t.getTaskId());
            ret.put(t.getProcessInstanceId(), task);
        }
        return ret;
    }

    @Override
    public Multimap<Long, TaskModel> getTaskModelsByProcessInstanceIdsAndStates(Collection<Long> processInstanceIds, Collection<State> states) {
        if (processInstanceIds.isEmpty()) return HashMultimap.create();
        List<String> taskNames = Lists.newArrayList(Collections2.transform(states, new Function<State, String>() {
            @Override
            public String apply(State state) {
                return state.getName();
            }
        }));
        Query query = processEm.createNamedQuery("TasksByProcessInstanceIdsAndTaskNames");
        query.setParameter("processInstanceIds", processInstanceIds);
        query.setParameter("taskNames", taskNames);
        List<TaskIdNameProcessInstanceIdTuple> tuples = query.getResultList();
        Multimap<Long, TaskModel> ret = LinkedListMultimap.create();
        for (TaskIdNameProcessInstanceIdTuple t : tuples) {
            TaskModel task = getTaskWithContent(t.getTaskId());
            ret.put(t.getProcessInstanceId(), task);
        }
        return ret;
    }

    @Override
    public Map<Long, Long> getProcessInstanceIdsByTaskIds(List<Long> taskIds) {
        if (taskIds.isEmpty()) return Collections.emptyMap();
        Query query = processEm.createNamedQuery("ProcessInstanceIdNamesByTaskIds");
        query.setParameter("taskIds", taskIds);
        List<TaskIdNameProcessInstanceIdTuple> tuples = query.getResultList();
        Map<Long, Long> ret = Maps.newHashMap();
        for (TaskIdNameProcessInstanceIdTuple t : tuples) {
            ret.put(t.getTaskId(), t.getProcessInstanceId());
        }
        return ret;
    }

    @Override
    public Map<String, Object> getContentById(long contentId) {
        Content content = taskClient.getContent(contentId);
        Object result = ContentMarshallerHelper.unmarshall(MARSHALLING_STRATEGY_TYPE, content.getContent(),
                getMarshallerContext(), null);
        return (Map<String, Object>) result;
    }

    @Override
    public Map<String, Object> getContentByTaskId(long taskId) {
        Task task = taskClient.getTask(taskId);
        return getContentById(task.getTaskData().getDocumentContentId());
    }

    @Override
    public List<TaskModel> listAllTasksByStates(Collection<State> states) {
        if (states.isEmpty()) return Collections.emptyList();
        List<String> taskNames = Lists.newArrayList(Collections2.transform(states, new Function<State, String>() {
            @Override
            public String apply(State state) {
                return state.getName();
            }
        }));
        Query query = processEm.createNamedQuery("AllTasksByTaskNames");
        query.setParameter("taskNames", taskNames);
        List<TaskSummary> taskSummaries = query.getResultList();
        List<TaskModel> taskModels = Lists.transform(taskSummaries, taskSummaryToModelFn);
        return Lists.newArrayList(taskModels);
    }

    private List<TaskModel> listAvailableTasksByUser(String userId, int first, int limit) {
        List<TaskSummary> taskSummaries = taskClient.getTasksAssignedAsPotentialOwner(
                userId, groups, "en-UK", first, limit);
        List<TaskModel> taskModels = Lists.transform(taskSummaries, taskSummaryToModelFn);
        return Lists.newArrayList(taskModels);
    }

    private List<TaskModel> listAvaiableTasksByUserAndState(String userId, State state, int first, int limit) {
        Query query = processEm.createNamedQuery("TasksAssignedAsPotentialOwnerByState");
        query.setParameter("userId", userId);
        query.setParameter("taskName", state.getName());
        query.setParameter("groups", groups);
        query.setFirstResult(first);
        query.setMaxResults(limit);
        List<TaskSummary> taskSummaries = query.getResultList();
        List<TaskModel> taskModels = Lists.transform(taskSummaries, taskSummaryToModelFn);
        return Lists.newArrayList(taskModels);
    }

    private List<TaskModel> listAllAvaiableTasks(int first, int limit) {
        List<TaskSummary> taskSummaries = (List<TaskSummary>) taskClient.query(
                entityMappings.get("TasksAssignedAsBusinessAdministrator").applyParams(
                        ImmutableMap.of("userId", "'Administrator'")), limit, first);
        List<TaskModel> taskModels = Lists.transform(taskSummaries, taskSummaryToModelFn);
        return Lists.newArrayList(taskModels);
    }

    private List<TaskModel> listAllAvaiableTasksByState(State state, int first, int limit) {
        List<TaskSummary> taskSummaries = (List<TaskSummary>) taskClient.query(
                entityMappings.get("TasksAssignedAsBusinessAdministratorByState").applyParams(
                        ImmutableMap.of("userId", "'Administrator'", "taskName", wrap(state.getName()))),
                limit, first);
        List<TaskModel> taskModels = Lists.transform(taskSummaries, taskSummaryToModelFn);
        return Lists.newArrayList(taskModels);
    }

    private TaskModel getTask(long taskId) {
        Task task = taskClient.getTask(taskId);
        TaskSummary taskSummary = taskToSummaryFn.apply(task);
        TaskModel taskModel = taskSummaryToModelFn.apply(taskSummary);
        return taskModel;
    }

    private ContentMarshallerContext getMarshallerContext() {
        return hornetQHTWorkItemHandler.getMarshallerContext();
    }

    private String wrap(String target) {
        return "'" + target + "'";
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Autowired
    @Qualifier("jbpmDataSource")
    public void setDS(DataSource dataSource) {
        setDataSource(dataSource);
    }
}
